package com.rickie.doris.streaming;

import com.rickie.doris.config.Constants;
import com.rickie.doris.domain.UserBehaviorEvent;
import com.rickie.doris.domain.UserBehaviorEventSchema;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.cfg.DorisSink;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;
import java.util.*;

/**
 * 处理用户访问记录 pvuv
 */
public class UserBehaviorToDoris {
    //定义用于标识旁路输出流的 lateDataTag
    private static final OutputTag<UserBehaviorEvent> lateDataTag = new OutputTag<UserBehaviorEvent>("late-data") {};

    public static void main(String[] args) throws Exception {
        //获取Flink 的运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置EventTime语义，从flink 1.12.0开始默认的时间语义为event-time。
        //env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //设置watermark生成周期  默认200ms，一般不去改
        env.getConfig().setAutoWatermarkInterval(200);

        //Checkpoint 配置，默认情况下，state会保存在taskmanager的内存中，checkpoint会存储在JobManager的内存中。
        //每隔5000ms启动一个检查点（设置CheckPoint的周期）
        env.enableCheckpointing(5000);
        //设置模式为.EXACTLY_ONCE（默认值），还可以设置为AT_LEAST_ONCE
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //checkpoint的最小间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        //检查点必须在1min内完成，或者被丢弃（checkPoint的超时时间）
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        //同一时间只允许执行一个检查点
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        //表示Flink处理程序被cancel后，会保留CheckPoint数据，以便根据实际需要恢复到指定的CheckPoint
        env.getCheckpointConfig().enableExternalizedCheckpoints(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        // Kafka参数
        String sourceTopic = Constants.TOPIC;
        Properties sourceProp = new Properties();
        sourceProp.setProperty("bootstrap.servers",Constants.KAFKA_BOOTSTRAP_SERVER);
        sourceProp.setProperty("group.id","test.group");

        FlinkKafkaConsumer<UserBehaviorEvent> myConsumer = new FlinkKafkaConsumer<>(
                sourceTopic, new UserBehaviorEventSchema(), sourceProp);
        //默认消费策略
        myConsumer.setStartFromGroupOffsets();

        // 设置watermark
        myConsumer.assignTimestampsAndWatermarks(
                WatermarkStrategy.forGenerator((ctx)-> new PeriodicWatermarkGenerator())
                        .withTimestampAssigner((ctx) -> new TimestampExtractor())
        );
        //设置Kafka数据源
        DataStreamSource<UserBehaviorEvent> userBehaviorStream = env.addSource(myConsumer);
        //处理流数据
        SingleOutputStreamOperator summaryStream = userBehaviorStream
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(60)))
                // 设置允许延迟的时间，默认为0
                .allowedLateness(Time.seconds(0))
                // 保存延迟数据
                .sideOutputLateData(lateDataTag)
                // 处理每个窗口内的所有元素
                .process(new ProcessAllWindowFunction<UserBehaviorEvent, Tuple3<String, Long, Integer>, TimeWindow>() {
                    /**
                     * 对一个窗口内的元素进行处理，窗口内的元素缓存在Iterable<IN>，进行处理后输出到Collector<OUT>中
                     * 我们可以输出一到多个结果
                     */
                    @Override
                    public void process(Context context, Iterable<UserBehaviorEvent> elements, Collector<Tuple3<String, Long, Integer>> out) {
                        Long pv = 0L;
                        Set<Integer> userIds = new HashSet<>();
                        Iterator<UserBehaviorEvent> iterator = elements.iterator();
                        while (iterator.hasNext()) {
                            UserBehaviorEvent userBehavior = iterator.next();
                            pv++;
                            userIds.add(userBehavior.getUserId());
                        }
                        // 测试输出，生产环境可以删除
                        TimeWindow window = context.window();
                        // 将当前窗口的起止时间整理成字符串输出，方便调试
                        String value = String.format("window, %s - %s, %d, %d",
                                // 当前窗口的起始时间
                                time(window.getStart()),
                                // 当前窗口的结束时间
                                time(window.getEnd()),
                                // pv
                                pv,
                                // uv
                                userIds.size());
                        System.out.println(value);
                        out.collect(new Tuple3<>(timeDate(window.getStart()), pv, userIds.size()));
                    }
                });

        // 侧路输出 - 把迟到的数据暂时打印到控制台，实际可以保存到其他存储介质中
        summaryStream.getSideOutput(lateDataTag).print("LateEvent -> ");

        SingleOutputStreamOperator dorisStream = summaryStream.map(new MapFunction<Tuple3<String, Long, Integer>, RowData>() {
            @Override
            public RowData map(Tuple3<String, Long, Integer> tuple3) throws Exception {
                GenericRowData genericRowData = new GenericRowData(3);
                genericRowData.setField(0, StringData.fromString(tuple3.getField(0)));
                genericRowData.setField(1, tuple3.getField(1));
                genericRowData.setField(2, tuple3.getField(2));
                return genericRowData;
            }
        });

        // -------- sink with RowData stream --------
        String[] fields = {"visitDate", "pv", "uv"};
        LogicalType[] types = {new VarCharType(), new BigIntType(), new IntType()};

        dorisStream.addSink(
                DorisSink.sink(
                        fields,
                        types,
                        DorisReadOptions.builder().build(),
                        DorisExecutionOptions.builder()
                                .setBatchSize(3)
                                .setBatchIntervalMs(0L)
                                .setMaxRetries(3)
                                .build(),
                        DorisOptions.builder()
                                .setFenodes("centos-103:8030")
                                .setTableIdentifier("example_db.pvuvsummary2")
                                .setUsername("root")
                                .setPassword("doris").build()
                )).name("doris-sink").setParallelism(1);

        // 打印输出
        dorisStream.print("SUMMARY: ").name("print").setParallelism(1);
        // execute
        env.execute("UserBehaviorToDoris");
    }

    /**
     * 将时间戳格式化输出
     * @param timestamp
     * @return
     */
    public static String time(long timestamp) {
        return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(timestamp));
    }

    public static String timeDate(long timestamp) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        return sdf.format(timestamp);
    }

    /**
     * 基于事件或者周期性地生成watermark
     */
    private static class PeriodicWatermarkGenerator implements WatermarkGenerator<UserBehaviorEvent> {
        private final long maxOutOfOrderness = 0; //2 * 1000 = 2秒
        private long currentMaxTimestamp;

        @Override
        public void onEvent(UserBehaviorEvent userBehaviorEvent, long eventTimestamp, WatermarkOutput watermarkOutput) {
            // 每来一条数据，将这条数据与currentMaxTimestamp比较，看是否需要更新watermark
            //System.out.println(eventTimestamp);
            currentMaxTimestamp = Math.max(currentMaxTimestamp, eventTimestamp);
        }

        /**
         * 周期性更新watermark 间隔时间看setAutoWatermarkInterval的参数
         * @param watermarkOutput
         */
        @Override
        public void onPeriodicEmit(WatermarkOutput watermarkOutput) {
            // 发出的watermark = 当前最大时间戳 - 最大乱序时间
            long effectiveWatermark = currentMaxTimestamp - maxOutOfOrderness;
            //System.out.println("Watermark: " + effectiveWatermark);
            watermarkOutput.emitWatermark(new Watermark(effectiveWatermark));
        }
    }

    /**
     * TimestampAssigner 负责从事件中提取时间戳字段
     */
    private static class TimestampExtractor implements TimestampAssigner<UserBehaviorEvent> {
        @Override
        public long extractTimestamp(UserBehaviorEvent userBehaviorEvent, long l) {
            //System.out.println(userBehaviorEvent.toString());
            return userBehaviorEvent.getTs();
        }
    }
}
